package com.tool4j.sse;

import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

@Component
public class SseEventPublisher {

    // 存储每个 client 的 Sink
    private final ConcurrentMap<String, Sinks.Many<String>> clients = new ConcurrentHashMap<>();

    /**
     * 获取某个客户端的 Flux
     */
    public Flux<String> getEventFlux(String clientId) {
        Sinks.Many<String> sink = clients.computeIfAbsent(clientId, id ->
                Sinks.many().multicast().onBackpressureBuffer()
        );
        return sink.asFlux();
    }

    /**
     * 向指定客户端发送消息
     */
    public void sendMessageToClient(String clientId, String message) {
        Sinks.Many<String> sink = clients.get(clientId);
        if (sink == null) {
            return;
        }
        Sinks.EmitResult result = sink.tryEmitNext(message);
        if (result.isFailure()) {
            System.err.println("推送失败: " + clientId + ", 原因: " + result);
            removeClient(clientId); // 统一清理
        }
    }

    /**
     * 向指定客户端发送消息并结束
     */
    public void sendComplete(String clientId) {
        sendMessageToClient(clientId, "event: stop");
    }

    /**
     * 显式移除客户端
     */
    public void removeClient(String clientId) {
        Sinks.Many<String> removed = clients.remove(clientId);
        if (removed != null) {
            // 可选：关闭 sink
            removed.tryEmitComplete(); // 通知所有订阅者完成
        }
    }
}